package loadData

import (
	"errors"
	"fmt"
	"github.com/panjf2000/ants/v2"
	"github.com/vbauerster/mpb/v8"
	cp "db2s/componentInit"
	"db2s/global"
	"db2s/inputArg"
	"db2s/outPut"
	quit "db2s/quitDispos"
	mq "db2s/topic-mq"
	"math/rand"
	"os"
	"reflect"
	"strings"
	"sync"
	"time"
)

var (
	monitorListener mq.MonitorListener
)

func GetCheckSchemaMetaData() *CheckSchemaMetaData {
	return &CheckSchemaMetaData{}
}
func getSchemaTableSeparation(s string) []string {
	return strings.Split(s, global.SchemaTableSplit)
}
func (t CheckSchemaMetaData) Check() {
	var (
		tIndex, tIndexNew       map[string]map[string]any
		tableList, tableListNew []string                        //task table
		tableSum                []string                        //sum table
		Tpods                   = make(map[string]*global.TPod) //pod init
		TDistributed            map[string]map[string]string
		TablePartition          map[string]global.PartitionGarth
	)
	tIndexNew = make(map[string]map[string]any)
	ChanSchemaContext := &SchemaInitContext{}
	for _, v := range t.TableMD {
		var metaDataColumn map[string]global.TableMetaGather
		//校验表初始化
		if t.ActiveSwitchV.TableStructInit {
			metaDataColumn, _ = v.StructInitToColumn(v.TableList)
		}
		//tables全局参数初始化
		if t.ActiveSwitchV.GlobalParameterInit {
			v.GlobalLike(v.TableList)
		}
		//索引信息初始化
		if t.ActiveSwitchV.TableIndexInit {
			tIndex, _ = v.IndexInitToColumn(v.TableList)
		}
		//集群shard初始化
		//if t.ActiveSwitchV.ClusterShardInit {
		//	TDistributed, _ = v.GreatdbTableDistributed(v.TableList)
		//}
		//分区表数据获取
		if t.ActiveSwitchV.PartitionInit {
			TablePartition, _ = v.GetTablePartition(v.TableList)
		}
		//校验表索引过滤（筛选无索引表）
		if t.ActiveSwitchV.TableIndexFilter {
			tableList = v.CheckIndexObject(v.TableList, tIndex)
		}
		//迁移对象优化
		//tableList = v.TableObjectSeqOptimizer(v.TableList, TDistributed["src"])

		//var continueTableList map[string]int
		//break point table list
		if getBreakPointStatus(GlobalPConfigs) {
			var s []string
			continueTableList := getBreakPointTableList()
			for _, r := range tableList {
				if _, ok := continueTableList[strings.Join(getSchemaTableSeparation(r), ".")]; !ok {
					s = append(s, r)
				}
			}
			tableList = s
		}
		for _, tableN := range tableList {
			s := strings.Split(tableN, global.SchemaTableSplit)
			schema := s[0]
			table := s[1]
			var indexType any
			var indexColumn any
			if v1, ok := tIndex[tableN]; ok {
				if v1["indexType"] == nil {
					indexType = "missIndex"
				} else {
					indexType = v1["indexType"]
				}
				if _, ok2 := v1["indexColumn"]; ok2 {
					if v1["indexColumn"] == nil {
						indexColumn = ""
						indexType = "missIndex"
					} else {
						indexColumn = strings.Join(v1["indexColumn"].([]string), ",")
					}
				}
			}
			tIndexNew[tableN] = tIndex[tableN]
			tableListNew = append(tableListNew, tableN)
			Tpods[fmt.Sprintf("%s.%s", schema, table)] = t.podInit(map[string]any{"schema": schema, "table": table, "indexType": indexType, "indexColumn": indexColumn, "cresult": "ok"})
		}
		ChanSchemaContext.ChanSchemaContext = append(ChanSchemaContext.ChanSchemaContext, &SchemaInitContext{
			TableList:     tableList,
			TableMiss:     v.TableMiss,
			TcolumnMeta:   metaDataColumn,
			TDistributed:  TDistributed,
			IndexData:     tIndex,
			TablePartiton: TablePartition,
		})
	}
	tableSum = append(tableSum, tableListNew...)
	ChanSchemaContext.TableSum = tableSum
	global.GetIoc().RegisterContext("terminalPods", Tpods)
	global.GetIoc().RegisterContext("schemaInitContext", ChanSchemaContext)
}

func (sp *SchedulePlan2) SchemaToTable() string {
	return fmt.Sprintf("%s.%s", sp.Object.Schema, sp.Object.Table)
}
func FuncExportVar() {
	//函数调用映射
	FuncRef = global.GetIoc().GetBean("checkObjectFuncContext").(map[string]reflect.Value)
	ScheduleTaskInitFuncRef = global.GetIoc().GetBean("scheduleTaskInitFuncContext").(map[string]reflect.Value)
	closeObject = global.GetIoc().GetBean("closeObject").(quit.Close)
	MqListenerSInitFunc = global.GetIoc().GetBean("mqListenerSInitFuncContext").(map[string]reflect.Value)
}

// ExportVar 导出变量 方便使用
func ExportVar() {
	//producer 生产者
	producer = global.GetIoc().GetBean("producer").(mq.Producer)
	//consumer 消费者
	consumer = global.GetIoc().GetBean("consumer").(mq.Consumer)
	//monitorListener mq 监控
	monitorListener = global.GetIoc().GetBean("monitorListener").(mq.MonitorListener)
	terminalPods = global.GetIoc().GetBean("terminalPods").(map[string]*global.TPod)
}

type sendTopicF func(check CheckMode, topic Topic, tableName string, msg map[string]any) bool
type sendEndDispos func()

func (context *SchedulePlanContext) GetIndexMode() CheckMode {
	c := global.GetIoc().GetBean("globalConfig").(*GlobalParametersConfig)
	switch c.rules.CheckMode {
	case "load":
		return "load"
	}
	return "load"
}
func getCheckMod() CheckMode {
	c := global.GetIoc().GetBean("globalConfig").(*GlobalParametersConfig)
	switch c.rules.CheckMode {
	case "load":
		return PlanContext.GetIndexMode()
	}
	return ""
}

func (context *SchedulePlanContext) objectEndMark(index CheckMode) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].ObjectEndMark()
}
func loadOutputToTopic(a *SchedulePlan2, table string, plan map[string]*SchedulePlan2, endMark bool) {
	var (
		topic  Topic
		logSeq = rand.NewSource(time.Now().UnixNano()).Int63()
	)
	if len(a.ReadOptimizer.IndexColumn) > 0 {
		topic = loadOutputIsSegmentQuery
	} else if len(a.ReadOptimizer.IndexColumn) == 0 {
		topic = loadOutputMissSegmentQuery
	}
	sendMsg(getCheckMod(), topic, table, map[string]any{"logSeq": logSeq, "segmentSql": map[string]string{}, "plan": plan, "subTask": a.subTaskInfo, "topic": string(topic)})
	if endMark {
		PlanContext.objectEndMark(getCheckMod())
	}
}
func loadInputToTopic(a *SchedulePlan2, table string, plan map[string]*SchedulePlan2, endMark bool) {
	var (
		topic  Topic
		logseq = rand.NewSource(time.Now().UnixNano()).Int63()
	)
	topic = loadInputFileData
	sendMsg(getCheckMod(), topic, table, map[string]any{"logSeq": logseq, "segmentSql": map[string]string{}, "plan": plan, "subTask": a.subTaskInfo, "topic": string(topic)})
	if endMark {
		PlanContext.objectEndMark(getCheckMod())
	}
}
func loadToTopic(a *SchedulePlan2, table string, k map[string]*SchedulePlan2, endMark bool) {
	if funcValue, ok := loadModeFunc()[strings.ToLower(GlobalPConfigs.rules.Load.LoadMode)]; ok {
		if funcValue.Kind() == reflect.Func {
			arguments := []reflect.Value{reflect.ValueOf(a),
				reflect.ValueOf(table),
				reflect.ValueOf(k),
				reflect.ValueOf(endMark)}
			funcValue.Call(arguments)
		}
	}
}
func loadModeFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"output": reflect.ValueOf(loadOutputToTopic),
		"input":  reflect.ValueOf(loadInputToTopic),
	}
}

func checkObjectFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"load": reflect.ValueOf(loadToTopic),
	}
}
func loadModeFileFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"input":  reflect.ValueOf(loadDataInputFileInit),
		"output": reflect.ValueOf(loadDataOutputFileInit),
	}
}
func sendScheduleMsgFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"load": reflect.ValueOf(initLoadSchedulePlan),
	}
}
func mqListenerSInitFunc() map[string]reflect.Value {
	return map[string]reflect.Value{
		"load": reflect.ValueOf(loadDataMQListenInitFunc),
	}
}

// NewSchedulePlanContext 创建执行计划上下文
func schedulePlanContext(config *SchedulePlanContextConfig) *SchedulePlanContext {
	var (
		event = "[schedulePlanContext]"
	)
	initContext := global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	if PlanContext == nil {
		lock.Lock()
		defer lock.Unlock()
		if PlanContext == nil {
			//初始化检查相关组件
			WLog.Debug(fmt.Sprintf("(%d) %s NewSchedulePlanContext global config init object is %v", 1, event, initContext.TableSum))
			global.GetIoc().RegisterContext("checkObjectFuncContext", checkObjectFunc())
			global.GetIoc().RegisterContext("scheduleTaskInitFuncContext", sendScheduleMsgFunc())
			global.GetIoc().RegisterContext("mqListenerSInitFuncContext", mqListenerSInitFunc())
			FuncExportVar()
			err := initContextBean(config)
			if err != nil {
				fmt.Println("创建执行计划上下文 NewSchedulePlanContext 失败", err)
				return nil
			}
			//初始化上下文线程池
			pool, err := ants.NewPool(len(initContext.TableSum))
			if err != nil {
				return nil
			}
			progress := mpb.New(mpb.WithWidth(80))
			global.GetIoc().RegisterContext("progress", progress)
			//初始化上下文
			PlanContext = &SchedulePlanContext{
				Pool:                      pool,
				AutoCloseMqComponentTable: cp.InitAutoCloseMqComponentTable(),
				BeginTime:                 config.BeginTime,
				TableList:                 initContext.TableSum,
			}
			PlanContext.TaskBarTable = cp.InitBarMonitor(cp.Bar{
				TaskTotal: int64(len(initContext.TableSum)),
				Task:      len(initContext.TableSum),
				CheckMod:  config.GtCheckConfig.SecondaryL.RulesV.CheckMode,
				Object:    config.GtCheckConfig.SecondaryL.SchemaV.CheckIndexObject,
				BeginTime: config.BeginTime,
				TtyBarFin: GlobalPConfigs.TtyBarFin,
			}, len(GetCheckToTopicMQListeners()[cp.CheckMode(config.GtCheckConfig.SecondaryL.RulesV.CheckMode)]))
			//注册ioc 容器
			global.GetIoc().RegisterContext("schedulePlanContext", PlanContext)
			//导出变量
			ExportVar()
			//初始化执行计划表
			PlanContext.ChanSchedulePlanTable, PlanContext.TableMiss = InitToSchedulePlan2Table()
			return PlanContext
		}
	}
	return PlanContext
}

// initContextBean 初始化上下文需要用到的组件
func initContextBean(config *SchedulePlanContextConfig) error {
	//TODO: checkConfig 行哥 你来写
	createMq := config.MQConfig.CreateMq(GetCheckToTopicMQListeners())
	if createMq == nil {
		return errors.New("初始化 mq 失败")
	}
	return nil
}

func (context *SchedulePlanContext) Close() {
	for _, component := range context.AutoCloseMqComponentTable {
		context.wg.Add(1)
		func() {
			defer context.wg.Done()
			component.Close(false)
		}()
	}
	context.wg.Wait()
}
func getBreakPointTableList() map[string]int {
	var (
		continueTableList = make(map[string]int)
	)
	ff := outPut.FileOut("text", map[string]any{"fileName": global.BreakPointFileName, "rowsS": "\n"})
	m := ff.Read1()
	for {
		select {
		case c, ok := <-m.(chan any):
			if !ok {
				return continueTableList
			}
			switch fmt.Sprintf("%v", reflect.TypeOf(c)) {
			case "*string":
				continueTableList[*c.(*string)]++
			}
		}
	}
}

// sendStart
func schedulePlanStart() {
	err := PlanContext.Pool.Submit(func() {
		//发送消息
		fmt.Println("-- db2s begin check table --")
		var (
			taskEndMark           = false
			subTaskSum            = 1
			ChanSchedulePlanTable = PlanContext.ChanSchedulePlanTable.([]map[string]*SchedulePlan2)
		)
		if len(ChanSchedulePlanTable) == 0 || ChanSchedulePlanTable == nil {
			os.Exit(1)
			return
		}
		for _, v := range PlanContext.TableMiss {
			terminalPods[v] = &global.TPod{TableMiss: true}
			var sheetName string
			for k, l := range GlobalPConfigs.result.XlsV.XlsSheetStructMap {
				if reflect.TypeOf(global.XStructMissTable{}) == l {
					sheetName = k
					break
				}
			}
			ResultStatisticsMq <- map[string]any{"sheet": sheetName,
				"data": strings.Join(strings.Split(v, "."), global.StructColumnSplit)}
		}
		for _, k := range ChanSchedulePlanTable {
			for table, a := range k {
				global.GTableCurry <- struct{}{}
				if subTaskSum == len(PlanContext.TableList) {
					taskEndMark = true
				}
				a.Object.TaskSeq = subTaskSum
				if funcName, ok := FuncRef[GlobalPConfigs.rules.CheckMode]; ok {
					if funcName.Kind() == reflect.Func {
						arguments := []reflect.Value{reflect.ValueOf(a),
							reflect.ValueOf(table),
							reflect.ValueOf(k),
							reflect.ValueOf(taskEndMark)}
						funcName.Call(arguments)
					}
				} else {
					fmt.Println("-- db2s report: checkObject parameter selection error, please refer to the log file for details, enable debug to get more information -- ")
					os.Exit(1)
				}
				subTaskSum++
			}
		}
	})
	if err != nil {
		return
	}
}

// 判断chan是否make
func chanNil(f chan any) bool {
	if f == nil || reflect.ValueOf(f).IsNil() {
		return true
	} else {
		return false
	}
}

func resultAsyncHronous(wg *sync.WaitGroup) {
	var activeDataFileTatilMap = map[string]any{
		"load": global.XLoadSheetTail{},
	}
	defer func() {
		wg.Done()
	}()
	var l []string
	for _, v := range global.StructNameJsonToSlice(activeDataFileTatilMap[GlobalPConfigs.rules.CheckMode]) {
		l = append(l, fmt.Sprintf("%v", v))
	}
	a := fmt.Sprintf("%v", strings.Join(l, " || "))
	if err := writeBreakPointData(GlobalPConfigs.rules.ActiveDataFin, []*string{&a}, 1); err != nil {
		//return
	}
	for {
		select {
		case m, ok := <-ResultStatisticsMq:
			var (
				sheetName string
				data      any
				xlsData   map[string]any
			)
			if !ok {
				if err := ResultStatisticsGeneral.Close(); err != nil {
					return
				}
				return
			}
			if v, ok1 := m.(map[string]any)["sheet"]; ok1 {
				sheetName = v.(string)
			}
			if v, ok1 := m.(map[string]any)["data"]; ok1 {
				if reflect.TypeOf(v).Kind() == reflect.Struct {
					data = strings.Join(global.StructPointerValueToSlice(v), global.StructColumnSplit)
				} else {
					data = v
				}
			}
			xlsData = map[string]any{
				"sheet": sheetName,
				"data":  data,
			}
			if !ResultStatisticsGeneral.ResultAsynchronousWriteFile(xlsData) {
				return
			}
		case c, ok := <-BarResultStatisticMq:
			if ok && c != nil {
				var activeDataString string
				if reflect.TypeOf(c).Kind() == reflect.Struct {
					activeDataString = strings.Join(global.StructJsonPointerValueToSlice(c), " || ")
				} else {
					activeDataString = fmt.Sprintf("%v", c)
					activeDataString = strings.ReplaceAll(activeDataString, "\n", ",")
				}
				if err := writeBreakPointData(GlobalPConfigs.rules.ActiveDataFin, []*string{&activeDataString}, 1); err != nil {
					//return
				}
				var sheetName string
				for k, v := range GlobalPConfigs.result.XlsV.XlsSheetStructMap {
					if reflect.TypeOf(c) == v {
						sheetName = k
						break
					}
				}
				ResultStatisticsMq <- map[string]any{"sheet": sheetName, "data": c}
			}
		}
	}
}

func (context *SchedulePlanContext) Check() {
	var wg sync.WaitGroup
	switch GlobalPConfigs.result.Teletypewriter {
	case "bar":
		context.TaskBarTable.BarMonitor()
	case "topic":
		monitorListener.TurnMonitor()
	default:
		context.TaskBarTable.BarMonitor()
	}
	wg.Add(1)
	if !chanNil(ResultStatisticsMq) {
		go resultAsyncHronous(&wg)
	} else {
		wg.Done()
	}
	closeObject.AbnormalExit()
	//执行计划开始
	schedulePlanStart()
	//关闭检测
	context.Close()
	//关闭消费者 阻塞
	consumer.Shutdown()
	//关闭生产者
	producer.Shutdown()
	close(BarResultStatisticMq)
	close(ResultStatisticsMq)
	wg.Wait()
	closeObject.CloseAction()
	fmt.Println("")
	//输出结果信息
	fmt.Println("Check end")
	WLog.Info("db2s check object {", GlobalPConfigs.rules.CheckMode, "} complete !!!")
	fmt.Println("** db2s Overview of results **")
	fmt.Println("Check time: ", fmt.Sprintf("%.2fs", time.Since(context.BeginTime).Seconds()), "(Seconds)")
	//outPut.CheckResultOut(fmt.Sprintf("%v", time.Since(context.BeginTime).Seconds()), GlobalPConfigs.rules.CheckMode, terminalPods)
}

// Audit input
func Audit(m *arg.ConfigParameter) {
	schemaContext(&SchemaInitContextConfig{
		GtCheckConfig: m,
	}).Check()
	schedulePlanContext(&SchedulePlanContextConfig{
		GtCheckConfig: *m,
		MQConfig:      cp.NewDefaultMq(),
		BeginTime:     m.BeginTime,
	}).Check()
}
